package com.small.nacos.client.naming.beat;

import com.alibaba.nacos.api.common.Constants;
import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.CommonParams;
import com.alibaba.nacos.api.naming.NamingResponseCode;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.utils.NamingUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.small.nacos.client.naming.net.NamingProxy;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author zhoujin
 * @Date 2022/10/15 16:34
 */
public class BeatReactor {


    private NamingProxy serverProxy;

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
    //是否发送过心跳信息
    private Boolean lightBeatEnabled = false;

    //心跳任务缓存
    public final Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<String, BeatInfo>();

    public BeatReactor(NamingProxy serverProxy, int threadCount) {
        this.serverProxy = serverProxy;
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(
                threadCount, runnable -> {
            Thread thread = new Thread(runnable);
            thread.setDaemon(true);
            thread.setName("com.small.nacos.naming.beat.sender");
            return thread;
        }
        );
    }

    public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
        //添加心跳任务
        String beatKey = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
        BeatInfo oldBeatInfo;
        //停止之前的心跳任务
        if ((oldBeatInfo = dom2Beat.put(beatKey, beatInfo)) != null) {
            oldBeatInfo.setStopped(true);
        }
        scheduledThreadPoolExecutor.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    }

    public BeatInfo buildBeatInfo(String groupedServiceName, Instance instance) {

        BeatInfo beatInfo = new BeatInfo();
        beatInfo.setServiceName(groupedServiceName);
        beatInfo.setIp(instance.getIp());
        beatInfo.setPort(instance.getPort());
        beatInfo.setCluster(instance.getClusterName());
        beatInfo.setWeight(instance.getWeight());
        beatInfo.setMetadata(instance.getMetadata());
        beatInfo.setScheduled(false);
        beatInfo.setPeriod(instance.getInstanceHeartBeatInterval());
        return beatInfo;

    }

    public BeatInfo buildBeatInfo(Instance instance) {
        return buildBeatInfo(instance.getServiceName(), instance);
    }

    public void removeBeatInfo(String groupedName, Instance instance) {
        String beatKey = buildKey(groupedName, instance.getIp(), instance.getPort());
        BeatInfo beatInfo = dom2Beat.remove(beatKey);
        if (beatInfo == null) {
            return;
        }
        beatInfo.setStopped(true);
    }

    public String buildKey(String serviceName, String ip, int port) {
        return serviceName + Constants.NAMING_INSTANCE_ID_SPLITTER + ip + Constants.NAMING_INSTANCE_ID_SPLITTER + port;
    }

    class BeatTask implements Runnable {

        private BeatInfo beatInfo;

        public BeatTask(BeatInfo beatInfo) {
            this.beatInfo = beatInfo;
        }

        @Override
        public void run() {

            if (beatInfo.isStopped()) {
                return;
            }
            long nextTime = beatInfo.getPeriod();

            try {
                JsonNode result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
                //获取下一次执行时间
                JsonNode clientBeatJsonNode = result.get("clientBeatInterval");
                long interval = clientBeatJsonNode == null ? 0 : clientBeatJsonNode.asLong();
                boolean lightBeatEnabled = false;
                if (result.has(CommonParams.LIGHT_BEAT_ENABLED)) {
                    lightBeatEnabled = result.get(CommonParams.LIGHT_BEAT_ENABLED).asBoolean();
                }
                BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
                if (interval > 0) {
                    nextTime = interval;
                }
                int code = NamingResponseCode.OK;
                if (result.has(CommonParams.CODE)) {
                    code = result.get(CommonParams.CODE).asInt();
                }
                if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
                    Instance instance = new Instance();
                    instance.setPort(beatInfo.getPort());
                    instance.setIp(beatInfo.getIp());
                    instance.setWeight(beatInfo.getWeight());
                    instance.setMetadata(beatInfo.getMetadata());
                    instance.setClusterName(beatInfo.getCluster());
                    instance.setServiceName(beatInfo.getServiceName());
                    instance.setInstanceId(instance.getInstanceId());
                    instance.setEphemeral(true);
                    try {
                        serverProxy.registerService(beatInfo.getServiceName(),
                                NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
                    } catch (Exception ignore) {
                    }
                }

            } catch (NacosException e) {
                e.printStackTrace();
            } finally {
                scheduledThreadPoolExecutor.schedule(new BeatTask(this.beatInfo), nextTime, TimeUnit.MILLISECONDS);
            }

        }
    }
}
